package com.ibm.cps.processors;

import java.util.Collection;
import java.util.List;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.RandomUtil;
import com.ibm.util.exception.CPSException;

public abstract class SchemaInheritiveProcessor extends BasicProcessor {
	private static final Logger logger = Logger.getLogger(SchemaInheritiveProcessor.class);
	private ITimeSeriesSchema outputSchema;
	protected ITimeSeriesSchema parentSchema;
	
	public SchemaInheritiveProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
	}
	
	private ITimeSeriesSchema createOutputSchema(String tenantId) throws CPSException{
		ArrayNode tags = parentSchema.getTags().deepCopy();
		List<NewField> newFields = defineFieldsToBeAdded();
		for (NewField field : newFields) {
			ObjectNode fieldNode = JSONUtility.newObjectNode();
			fieldNode.put(MessageFields.NAME, field.fieldName);
			fieldNode.put(MessageFields.DATATYPE, field.dataType);
			tags.add(fieldNode);
		}
		String schemaid = RandomUtil.getRandomId();
		return new TimeSeriesSchema(tenantId, schemaid, tags, 1);
	}

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message)
			throws CPSException {
		Collection<ParsedDataMessage> returnedMessages = null;
		try{
			returnedMessages = runProcessorLogic(parentSchema, outputSchema, message);
			return returnedMessages;
		}catch(Exception logicException){
			logger.error("Exception occured in processor: " + this.getClass().getName() + " input: " + message);
			throw new CPSException(ErrorCode.NULL, logicException.getMessage(), logicException);
		}
	}
	
	
	
	protected abstract Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException;
	protected abstract List<NewField> defineFieldsToBeAdded();
	
	protected static class NewField {
		private final String fieldName;
		private final String dataType;
		
		public NewField(String fieldName, String dataType) {
			this.fieldName = fieldName;
			this.dataType = dataType;
		}

		@Override
		public String toString() {
			return "{\"Field\":{\"fieldName\":\"" + fieldName
					+ "\", \"dataType\":\"" + dataType + "\"}}";
		}
		
	}
	

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() throws CPSException {
			
		return outputSchema;
	}

	@Override
	public final void open() throws CPSException {
		initialize();
		outputSchema = createOutputSchema(tenantId);
	}

	public abstract void initialize() throws CPSException;
	
}
